# -*- coding: utf-8 -*-
# @Time    : 2025/10/18
# @Author  : 楚江涛
# @FileName: orchestrator_service.py
# @Software: PyCharm


"""
app/application/orchestrator_service.py
---------------------------------------
编排服务（Use-Case 层）—— 作为 API 与执行器之间的唯一入口。

职责：
- 接收启动/控制指令，装配并驱动 WorkflowExecutor；
- 订阅执行过程事件：同时更新 StateStore（快照）并转发到 EventPublisher；
- 维护 exec_id -> executor 的运行中映射，提供控制与查询；
- 封装与事件循环的交互（创建后台任务、异常兜底）。

依赖（通过构造函数注入）：
- events: EventPublisher（SSE/消息总线等）
- state:  StateStore（快照存储，用于 /status）
- runner: CodeRunner（代码节点执行器）

注意：
- 本实现为单进程内多执行并发；多实例部署时请将 StateStore/事件总线换成共享实现（如 Redis/DB）。
"""

from __future__ import annotations

import asyncio
import uuid
from typing import Dict, Any, Optional, Callable

from domain.enums import EventType, ExecStatus
from domain.errors import ExecNotFound, InvalidWorkflow
from execution.executor import WorkflowExecutor
from execution.interfaces import EventPublisher, StateStore, CodeRunner
from infrastructure.bgloop import bg_loop


class OrchestratorService:
    """
    OrchestratorService 将“外部世界”的调用（API 层）与执行器绑定起来，
    对外暴露面向用例的简洁方法：start/cancel/pause/resume/cancel_node/snapshot。
    """

    def __init__(
        self,
        events: EventPublisher,
        state: StateStore,
        runner: CodeRunner,
        *,
        max_parallelism: int = 8,
        node_timeout: Optional[float] = 600.0,
    ) -> None:
        self._events = events
        self._state = state
        self._runner = runner
        self._max_parallelism = max_parallelism
        self._node_timeout = node_timeout

        # 运行中的执行器映射
        self._executors: Dict[str, WorkflowExecutor] = {}

    # ======================================================================
    # 启动
    # ======================================================================

    def start(
        self,
        workflow: Dict[str, Any],
        *,
        context: Optional[Dict[str, Any]] = None,
        exec_id: Optional[str] = None,
    ) -> str:
        """
        启动一次执行：
        - 校验最基本的 DAG 结构；
        - 初始化 StateStore；
        - 创建并保存执行器实例；
        - 将执行器 run() 安排为后台任务；
        - 返回 exec_id。
        """
        self._validate_workflow_minimal(workflow)

        eid = exec_id or str(uuid.uuid4())
        self._state.init(eid)

        # 事件桥：执行器 → 状态快照 & 事件总线
        def emit(event: EventType, payload: Dict[str, Any]) -> None:
            self._apply_state_update(eid, event, payload)   # 更新快照
            self._events.publish(eid, event, payload)        # 转发事件

        # 构造执行器
        executor = WorkflowExecutor(
            exec_id=eid,
            wf=workflow,
            events=self._EventEmitterProxy(emit),  # 统一 EventType 类型
            runner=self._runner,
            max_parallelism=self._max_parallelism,
            node_timeout=self._node_timeout,
        )
        self._executors[eid] = executor

        # 启动后台任务
        async def _runner():
            try:
                await executor.run(initial_context=context or {})
            finally:
                # 执行结束，从映射中移除
                self._executors.pop(eid, None)
        try:
            loop = asyncio.get_running_loop()
            loop.create_task(_runner())
        except RuntimeError:
            bg_loop.submit(_runner())
        return eid

    # ======================================================================
    # 控制（全局/节点级）
    # ======================================================================

    def cancel(self, exec_id: str) -> None:
        ex = self._get(exec_id)
        ex.request_cancel()

    def pause(self, exec_id: str) -> None:
        ex = self._get(exec_id)
        ex.request_pause()

    def resume(self, exec_id: str) -> None:
        ex = self._get(exec_id)
        ex.request_resume()

    def cancel_node(self, exec_id: str, node_id: str) -> None:
        ex = self._get(exec_id)
        ex.cancel_node(node_id)

    # ======================================================================
    # 查询
    # ======================================================================

    def snapshot(self, exec_id: str) -> Dict[str, Any]:
        """
        返回状态快照（直接透传 StateStore）。
        若 exec 不存在但状态仍在 Store，可照常返回（例如执行刚结束）。
        """
        return self._state.snapshot(exec_id)

    def is_active(self, exec_id: str) -> bool:
        """判断执行是否仍在当前进程中运行"""
        return exec_id in self._executors

    def list_active(self) -> Dict[str, ExecStatus]:
        """
        简单枚举当前进程中活跃的执行（仅供观测/管理面使用）。
        注：状态以 StateStore 为准。
        """
        out: Dict[str, ExecStatus] = {}
        for eid in list(self._executors.keys()):
            snap = self._state.snapshot(eid)
            out[eid] = ExecStatus(snap.get("status", ExecStatus.PENDING.value))
        return out

    # ======================================================================
    # 内部：事件 → 状态快照
    # ======================================================================

    def _apply_state_update(self, exec_id: str, event: EventType, payload: Dict[str, Any]) -> None:
        """
        将执行器发出的事件映射为 StateStore 的快照更新。
        仅处理快照所需的最小集；业务日志/指标另行处理。
        """
        if event == EventType.EXEC_STARTED:
            self._state.update_status(exec_id, ExecStatus.RUNNING)

        elif event == EventType.NODE_STARTED:
            nid = payload.get("node_id", "")
            if nid:
                self._state.mark_node_running(exec_id, nid)

        elif event == EventType.NODE_FINISHED:
            nid = payload.get("node_id", "")
            if nid:
                self._state.mark_node_finished(exec_id, nid)

        elif event == EventType.NODE_FAILED:
            nid = payload.get("node_id", "")
            err = payload.get("error", "节点执行失败。")
            if nid:
                self._state.mark_node_failed(exec_id, nid, err)

        elif event == EventType.NODE_CANCELLED:
            nid = payload.get("node_id", "")
            if nid:
                # 对快照而言，取消也视作“结束”
                self._state.mark_node_finished(exec_id, nid)

        elif event == EventType.EXEC_FINISHED:
            status_str = payload.get("status", ExecStatus.SUCCEEDED.value)
            self._state.update_status(exec_id, ExecStatus(status_str))

        # 其他事件（日志、暂停/继续/准备取消）不强制写入快照
        # 可按需要补充 progress / metadata 等

    # ======================================================================
    # 内部：工具
    # ======================================================================

    def _get(self, exec_id: str) -> WorkflowExecutor:
        ex = self._executors.get(exec_id)
        if not ex:
            # 可能是：exec 不存在/已结束/不在本实例上
            raise ExecNotFound(f"执行实例不存在或已结束：exec_id={exec_id}")
        return ex

    @staticmethod
    def _validate_workflow_minimal(workflow: Dict[str, Any]) -> None:
        """
        基础校验（最小化）：nodes/edges 是否存在，节点 ID 唯一性，边引用是否存在。
        更严格的 schema 校验可在更上层做（Pydantic/JSONSchema）。
        """
        try:
            nodes = workflow["nodes"]
            edges = workflow.get("edges", [])
        except Exception as _:
            raise InvalidWorkflow("工作流缺少必要字段（nodes/edges）。")

        if not isinstance(nodes, list) or not nodes:
            raise InvalidWorkflow("工作流节点列表为空。")

        # 唯一性
        ids = [n["id"] for n in nodes if isinstance(n, dict) and "id" in n]
        if len(ids) != len(set(ids)):
            raise InvalidWorkflow("存在重复的节点 ID。")

        # 引用合法性
        idset = set(ids)
        for e in edges:
            s = e.get("sourceNodeID")
            t = e.get("targetNodeID")
            if s not in idset or t not in idset:
                raise InvalidWorkflow(f"连线引用了不存在的节点：{s} -> {t}。")

    # ======================================================================
    # 事件发射器代理（将调用方的 (EventType, payload) 适配到 Executor）
    # ======================================================================

    class _EventEmitterProxy:
        """
        执行器需要的是一个“有 publish(exec_id,event,payload)”接口的对象，
        但我们在 Orchestrator 中只想传入回调函数 emit(EventType, payload)。
        通过这个小代理把调用转换为 emit(event, payload)。
        """

        def __init__(self, emit: Callable[[EventType, Dict[str, Any]], None]) -> None:
            self._emit = emit

        # 让 executor.events.publish(self.exec_id, event, payload) 生效
        def publish(self, exec_id: str, event: EventType, payload: Dict[str, Any]) -> None:
            # exec_id 已包含在 payload 中（executor 内部已注入），这里只是透传
            self._emit(event, payload)

